package com.aotain.kafka;

import java.lang.reflect.Constructor;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringEncoder;

public class KafkaConsumer {
	private static String TOPIC = "importlog";
    private static final int THREAD_AMOUNT = 1;
    ConsumerConnector consumer = null;
    ExecutorService executor = null; 
    
    Properties props = new Properties();
    
    public KafkaConsumer(String zkserver,String topic)
	{
		
    	props.put("zookeeper.connect", zkserver);
    	props.put("group.id", "group1");
    	props.put("zookeeper.session.timeout.ms", "400");
    	props.put("zookeeper.sync.time.ms", "200");
    	props.put("auto.commit.interval.ms", "1000");;
    	props.put("serializer.class", StringEncoder.class.getName());	
    	props.put("auto.offset.reset", "smallest");
		TOPIC = topic;
	}
    
    /**
     * 启动消费者
     */
    public void StartConsumer(Class <? extends AbstractHanldMsgThread> hanldThread)
    {
    	 Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
         //每个topic使用多少个kafkastream读取, 多个consumer
         topicCountMap.put(TOPIC, THREAD_AMOUNT);
         //可以读取多个topic
//       topicCountMap.put(TOPIC2, 1);
         consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
         Map<String, List<KafkaStream<byte[], byte[]>>> msgStreams = consumer.createMessageStreams(topicCountMap );
         List<KafkaStream<byte[], byte[]>> msgStreamList = msgStreams.get(TOPIC);
          
         
         
         //BatcherBase base = (BatcherBase) con.newInstance(params);  //BatcherBase 为自定义类
         
         
         //使用ExecutorService来调度线程
         executor = Executors.newFixedThreadPool(THREAD_AMOUNT);
         for (int i = 0; i < msgStreamList.size(); i++) {
             KafkaStream<byte[], byte[]> kafkaStream = msgStreamList.get(i);
             
             try
             {
	             Class cls = Class.forName(hanldThread.getName());
	             Class[] paramTypes = { KafkaStream.class, int.class};
	             Object[] params = {kafkaStream, i}; // 方法传入的参数
	             Constructor con = cls.getConstructor(paramTypes);     //主要就是这句了
	             
	             executor.submit((AbstractHanldMsgThread)con.newInstance(params));
             //new HanldMessageThread(kafkaStream, i)
             }
             catch(Exception ex)
             {
            	 System.out.println(ex);
             }
         }
    }
    
    public void StopConsumer()
    {
    	if (executor != null) {
			  Logger.getLogger(KafkaConsumer.class).info("executor Shutdown");
			  executor.shutdown();
		  }
		  try {
		      if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) {
		          System.out.println("Timed out waiting for consumer threads to shut down, exiting uncleanly");
		      }
		  } catch (InterruptedException e) {
		      System.out.println("Interrupted during shutdown, exiting uncleanly");
		  }
		  
		  if (consumer != null) {
			  Logger.getLogger(KafkaConsumer.class).info("Consumer Shutdown");
			  consumer.shutdown();
		  }
		  
    } 
    
    public boolean IsStop(){
    	
    	if(executor!=null)
    		return executor.isShutdown();
    	return true;
    }
}
 
/**
 * 具体处理message的线程
 * @author Administrator
 *
 */
class HanldMessageThread extends AbstractHanldMsgThread {
 
    public HanldMessageThread(KafkaStream<byte[], byte[]> kafkaStream, int num) {
		super(kafkaStream, num);
		// TODO Auto-generated constructor stub
	}

	private KafkaStream<byte[], byte[]> kafkaStream = null;
    private int num = 0;
     
//    public void run() {
//        ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
//        while(iterator.hasNext()) {
//            String message = new String(iterator.next().message());
//            System.out.println("Thread no: " + num + ", message: " + message);
//        }
//    }

	@Override
	public void ExectueMessage(String message) {
		// TODO Auto-generated method stub
        //System.out.println("Thread no: " + num + ", message: " + message);
	}
}
